package com.yyx.up.analysis.tool;

import com.yyx.up.analysis.io.MySQLRedisTextOutPutFormat;
import com.yyx.up.analysis.mapper.AnalysisTextMapper;
import com.yyx.up.analysis.reducer.AnalysisTextReducer;
import com.yyx.up.common.constant.Names;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobStatus;
import org.apache.hadoop.util.Tool;

/**
 * Description: 分析数据的工具类
 * @Author: yyx
 * @Create: 2023/5/5 18:51
 * @Version: 8
 */
public class AnalysisTextTool implements Tool {
    @Override
    public int run(String[] strings) throws Exception {
        Job job = Job.getInstance();//获取job
        job.setJarByClass(AnalysisTextTool.class);//获取jar包路径

        Scan scan = new Scan();//利用扫描列族仅仅扫描观影数据，不扫描重复数据
        scan.addFamily(Bytes.toBytes(Names.UP_USER_Film.getValue()));

        // mapper
        TableMapReduceUtil.initTableMapperJob(
                Names.TABLE.getValue(),
                scan,
                AnalysisTextMapper.class,
                Text.class,
                Text.class,
                job
        );

        // reducer
        job.setReducerClass(AnalysisTextReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        // outputformat
        job.setOutputFormatClass(MySQLRedisTextOutPutFormat.class);

        boolean flg = job.waitForCompletion(true);
        if(flg){
            return JobStatus.State.SUCCEEDED.getValue();
        } else{
            return JobStatus.State.FAILED.getValue();
        }
    }

    @Override
    public void setConf(Configuration configuration) {

    }

    @Override
    public Configuration getConf() {
        return null;
    }
}
